package tt.dz.biz.service.rpc;

import java.io.InputStream;
import java.net.InetAddress;
import java.net.InetSocketAddress;
import java.nio.ByteBuffer;
import java.util.List;
import java.util.concurrent.Executor;
import java.util.concurrent.Executors;
import javax.annotation.PostConstruct;
import javax.annotation.PreDestroy;
import org.apache.avro.Protocol;
import org.apache.avro.io.BinaryEncoder;
import org.apache.avro.io.EncoderFactory;
import org.apache.avro.ipc.NettyServer;
import org.apache.avro.ipc.NettyTransportCodec;
import org.apache.avro.ipc.NettyTransportCodec.NettyDataPack;
import org.apache.avro.ipc.Responder;
import org.apache.avro.ipc.Server;
import org.apache.avro.util.ByteBufferOutputStream;
import org.jboss.netty.channel.ChannelEvent;
import org.jboss.netty.channel.ChannelFactory;
import org.jboss.netty.channel.ChannelHandlerContext;
import org.jboss.netty.channel.MessageEvent;
import org.jboss.netty.channel.socket.nio.NioServerSocketChannelFactory;
import org.jboss.netty.handler.execution.ExecutionHandler;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.stereotype.Service;
import tt.dz.common.LogUtil;

@Service("mainServer")
public class MainServer
{
  @Value("${ms.rpc.serverPort}")
  private int port;
  private static Server server;
  
  public void setPort(int port)
  {
    this.port = port;
  }
  
  @PostConstruct
  public void init()
  {
    try
    {
      startService();
    }
    catch (Exception e)
    {
      LogUtil.error("主服务器接口服务初始化失败", e);
    }
  }
  
  public void startService()
    throws Exception
  {
    try
    {
      InputStream in = getClass().getClassLoader().getResourceAsStream("msAvro.json");
      Protocol protocol = Protocol.parse(in);
      closeService();
      Responder responder = new RPCInterfaceService(protocol);
      Executor bossExecutor = Executors.newCachedThreadPool();
      Executor workerExecutor = Executors.newCachedThreadPool();
      ExecutionHandler executionHandler = new ExecutionHandler(workerExecutor)
      {
        public void handleUpstream(ChannelHandlerContext context, ChannelEvent e)
          throws Exception
        {
          if ((e instanceof MessageEvent))
          {
            NettyTransportCodec.NettyDataPack dataPack = (NettyTransportCodec.NettyDataPack)((MessageEvent)e).getMessage();
            List<ByteBuffer> req = dataPack.getDatas();
            ByteBufferOutputStream bbo = new ByteBufferOutputStream();
            BinaryEncoder out = EncoderFactory.get().binaryEncoder(bbo, null);
            bbo.prepend(req);
            InetSocketAddress remoteAddress = (InetSocketAddress)((MessageEvent)e).getRemoteAddress();
            out.writeString(remoteAddress.getAddress().getHostAddress());
            out.flush();
            dataPack.setDatas(bbo.getBufferList());
            super.handleUpstream(context, e);
          }
          else
          {
            super.handleUpstream(context, e);
          }
        }
      };
      ChannelFactory channelFactory = new NioServerSocketChannelFactory(bossExecutor, workerExecutor);
      server = new NettyServer(responder, new InetSocketAddress(this.port), channelFactory, executionHandler);
      server.start();
      LogUtil.debug("主服务器rpc接口服务启动完成，端口为：" + this.port);
    }
    catch (Exception e)
    {
      LogUtil.error("主服务器接口服务启动失败", e);
      throw e;
    }
    Protocol protocol;
  }
  
  @PreDestroy
  public void closeService()
    throws Exception
  {
    try
    {
      if (server != null) {
        server.close();
      }
      LogUtil.debug("主服务器rpc接口服务关闭成功");
    }
    catch (Exception e)
    {
      LogUtil.error("主服务器rpc接口服务关闭失败", e);
      throw e;
    }
  }
}
